#ifndef __M_BROKER_H__
#define __M_BROKER_H__
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"

#include "mq_connection.hpp"
#include "mq_consumer.hpp"
#include "mq_host.hpp"
#include "../mqcommon/mq_threadpool.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_logger.hpp"

namespace castlemq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using openChannelRequestPtr=std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr=std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr=std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr=std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr=std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr=std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr=std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr=std::shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr=std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr=std::shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr=std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr=std::shared_ptr<basicCancelRequest>;

    #define DBFILE "/meta.db"
    #define HOSTNAME "MyVirtualHost"
    class BrokerServer
    {
        public:
        typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
    
        BrokerServer(int port,const std::string& basedir)
        :_server(&_baseloop,muduo::net::InetAddress("0.0.0.0",port),
        "Server",muduo::net::TcpServer::kReusePort),
        _dispatcher(std::bind(&BrokerServer::onUnknownMessage, this, 
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
        _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher,
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
        _vitural_host(std::make_shared<VirtualHost>(HOSTNAME,basedir,basedir+DBFILE)),
        _consumer_manager(std::make_shared<ConsumerManager>()),
        _connection_manager(std::make_shared<ConnectionManager>()),
        _threadpool(std::make_shared<threadpool>())
        {
            //针对历史消息中的所有队列，初始化队列的消费者管理结构
            QueueMap qm=_vitural_host->allQueues();
            for(auto queue:qm)
            {
                _consumer_manager->initQueueConsumer(queue.first);
            }
            //注册业务处理函数
            _dispatcher.registerMessageCallback<castlemq::openChannelRequest>(std::bind(&BrokerServer::onOpenChannel,this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<castlemq::closeChannelRequest>(std::bind(&BrokerServer::onCloseChannel,this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<castlemq::declareExchangeRequest>(std::bind(&BrokerServer::onDeclareExchange,this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<castlemq::deleteExchangeRequest>(std::bind(&BrokerServer::onDeleteExchange,this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<castlemq::declareQueueRequest>(std::bind(&BrokerServer::onDeclareQueue,this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<castlemq::deleteQueueRequest>(std::bind(&BrokerServer::onDeleteQueue,this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<castlemq::queueBindRequest>(std::bind(&BrokerServer::onQueueBind,this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<castlemq::queueUnBindRequest>(std::bind(&BrokerServer::onQueueUnBind,this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<castlemq::basicPublishRequest>(std::bind(&BrokerServer::onBasicPublish,this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<castlemq::basicAckRequest>(std::bind(&BrokerServer::onBasicAck,this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<castlemq::basicConsumeRequest>(std::bind(&BrokerServer::onBasicConsume,this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<castlemq::basicCancelRequest>(std::bind(&BrokerServer::onBasicCancelConsume,this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                
            _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), 
            std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _server.setConnectionCallback(std::bind(&BrokerServer::onConnect, this, std::placeholders::_1));
        }
    
        void start()
        {
            _server.start();
            _baseloop.loop();
        }
    
    private:
        //打开信道
        void onOpenChannel(const muduo::net::TcpConnectionPtr& conn,
                            const openChannelRequestPtr& message,
                            muduo::Timestamp)
        {
            Connection::ptr mconn=_connection_manager->getConnection(conn);
            if(mconn.get()==nullptr)
            {
                DLOG("打开信道时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            return mconn->openChannel(message);
        }

        //关闭信道
        void onCloseChannel(const muduo::net::TcpConnectionPtr& conn,
            const closeChannelRequestPtr& message,
            muduo::Timestamp)
        {
            Connection::ptr mconn=_connection_manager->getConnection(conn);
            if(mconn.get()==nullptr)
            {
                DLOG("打开信道时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            return mconn->closeChannel(message);
        }

        //声明交换机
        void onDeclareExchange(const muduo::net::TcpConnectionPtr& conn,
            const declareExchangeRequestPtr& message,
            muduo::Timestamp)
        {
            Connection::ptr mconn=_connection_manager->getConnection(conn);
            if(mconn.get()==nullptr)
            {
                DLOG("声明交换机时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get()==nullptr)
            {
                DLOG("声明交换机时，没有找到信道!");
                return;
            }
            return cp->declareExchange(message);
        }

        //删除交换机
        void onDeleteExchange(const muduo::net::TcpConnectionPtr& conn,
            const deleteExchangeRequestPtr& message,
            muduo::Timestamp)
        {
            Connection::ptr mconn=_connection_manager->getConnection(conn);
            if(mconn.get()==nullptr)
            {
                DLOG("删除交换机时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get()==nullptr)
            {
                DLOG("删除交换机时，没有找到信道!");
                return;
            }
            return cp->deleteExchange(message);
        }
    //声明队列
        
        void onDeclareQueue(const muduo::net::TcpConnectionPtr& conn,
            const declareQueueRequestPtr& message,
            muduo::Timestamp)
        {
            Connection::ptr mconn=_connection_manager->getConnection(conn);
            if(mconn.get()==nullptr)
            {
                DLOG("声明队列时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get()==nullptr)
            {
                DLOG("声明队列时，没有找到信道!");
                return;
            }
            return cp->declareQueue(message);
        }
    //删除队列
        
        void onDeleteQueue(const muduo::net::TcpConnectionPtr& conn,
            const deleteQueueRequestPtr& message,
            muduo::Timestamp)
        {
            Connection::ptr mconn=_connection_manager->getConnection(conn);
            if(mconn.get()==nullptr)
            {
                DLOG("删除队列时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get()==nullptr)
            {
                DLOG("删除队列时，没有找到信道!");
                return;
            }
            return cp->deleteQueue(message);
        }
        //队列绑定
        void onQueueBind(const muduo::net::TcpConnectionPtr& conn,
            const queueBindRequestPtr& message,
            muduo::Timestamp)
        {
            Connection::ptr mconn=_connection_manager->getConnection(conn);
            if(mconn.get()==nullptr)
            {
                DLOG("绑定队列时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get()==nullptr)
            {
                DLOG("绑定队列时，没有找到信道!");
                return;
            }
            return cp->queueBind(message);
        }
        //队列解绑
        void onQueueUnBind(const muduo::net::TcpConnectionPtr& conn,
            const queueUnBindRequestPtr& message,
            muduo::Timestamp)
        {
            Connection::ptr mconn=_connection_manager->getConnection(conn);
            if(mconn.get()==nullptr)
            {
                DLOG("解绑队列时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get()==nullptr)
            {
                DLOG("解绑队列时，没有找到信道!");
                return;
            }
            return cp->queueUnbind(message);
        }
        //消息发布
        void onBasicPublish(const muduo::net::TcpConnectionPtr& conn,
            const basicPublishRequestPtr& message,
            muduo::Timestamp)
        {
            Connection::ptr mconn=_connection_manager->getConnection(conn);
            if(mconn.get()==nullptr)
            {
                DLOG("发布队列消息时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get()==nullptr)
            {
                DLOG("发布队列消息时，没有找到信道!");
                return;
            }
            return cp->basicPublish(message);
        }
        //消息确认
        void onBasicAck(const muduo::net::TcpConnectionPtr& conn,
            const basicAckRequestPtr& message,
            muduo::Timestamp)
        {
            Connection::ptr mconn=_connection_manager->getConnection(conn);
            if(mconn.get()==nullptr)
            {
                DLOG("队列消息确认时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get()==nullptr)
            {
                DLOG("队列消息确认时，没有找到信道!");
                return;
            }
            return cp->basicAck(message);
        }
        //队列消息订阅
        void onBasicConsume(const muduo::net::TcpConnectionPtr& conn,
            const basicConsumeRequestPtr message,
            muduo::Timestamp)
        {
            Connection::ptr mconn=_connection_manager->getConnection(conn);
            if(mconn.get()==nullptr)
            {
                DLOG("队列消息订阅时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get()==nullptr)
            {
                DLOG("队列消息订阅时，没有找到信道!");
                return;
            }
            return cp->basicConsume(message);
        }
        //队列消息取消订阅
        void onBasicCancelConsume(const muduo::net::TcpConnectionPtr& conn,
            const basicCancelRequestPtr& message,
            muduo::Timestamp)
        {
            Connection::ptr mconn=_connection_manager->getConnection(conn);
            if(mconn.get()==nullptr)
            {
                DLOG("取消队列消息订阅时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get()==nullptr)
            {
                DLOG("取消队列消息订阅时，没有找到信道!");
                return;
            }
            return cp->basicCancel(message);
        }
    
        
        void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn,
                            const MessagePtr& message,
                            muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }
    
        void onConnect(const muduo::net::TcpConnectionPtr& conn)
        {
            if(conn->connected())
            {
                _connection_manager->newConnection(conn,_codec,_consumer_manager,_vitural_host,_threadpool);
            }
            else
            {
                _connection_manager->deleteConnection(conn);
            }
        }
    private:   
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server;//服务器对象
        ProtobufDispatcher _dispatcher;//请求分发器对象，要向其中注册请求处理函数
        ProtobufCodecPtr _codec;//protobuf协议处理器,针对收到的请求数据进行protobuf协议处理
        VirtualHost::ptr _vitural_host;
        ConsumerManager::ptr _consumer_manager;
        ConnectionManager::ptr _connection_manager;
        threadpool::ptr _threadpool;
    };
}
#endif
